package trouger.dmt;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;

class DumpTableMapperEntity
implements MapperEntity<NullWritable, Text>{

	protected FieldIndexPair[] resultFieldsIndex;
	protected String separator;
	
	public void configure(String outputSeparator, FieldIndexPair[] resultFieldsIndex){
		this.separator = outputSeparator;
		this.resultFieldsIndex = resultFieldsIndex;
	}
	
	StringBuffer strbuf = new StringBuffer();
	NullWritable outKey = NullWritable.get();
	Text outValue = new Text();
	
	@Override
	public boolean acceptLine(int sourceIndex, String[] segment) {
		int i = 0;
		for (FieldIndexPair field : resultFieldsIndex){
			if (i++ > 0) strbuf.append(separator);
			strbuf.append(segment[field.getColumnIndex()]);
		}
		outValue.set(strbuf.toString());
		strbuf.delete(0, strbuf.length());
		return true;
	}
	@Override
	public NullWritable getKey() {
		return outKey;
	}
	@Override
	public Text getValue() {
		return outValue;
	}
	
}

class DumpTableMapper
extends BaseMapper<DumpTableMapperEntity, NullWritable, Text>{
	
	
	@Override
	public void configure(JobConf conf){
		super.configure(conf);
		
		FieldIndexPair[] resultFieldsIndex = BaseMapredUtils.getResultFieldIndices(conf, sourceCount);
		if (resultFieldsIndex == null){
			throw new IllegalArgumentException("result-fields-indices must be specified for a dump-table-runner");
		}
		
		String separator = BaseMapredUtils.getConfigString(conf, BaseMapredParams.SEPARATOR_PARAM_NAME);
		if (separator == null){
			separator = "\t";
		}
		
		this.mapperEntity.configure(separator, resultFieldsIndex);
	}
}

/**
 * This Runner class is used to do single-table dump operation.
 * You can extract fields from the single source table, modify field value with the help of SourceRecordFilter, and so on. 
 */
public class DumpTableRunner 
extends BaseMapredRunner<DumpTableMapper, DumpTableTrivialReducer> {
	@Override 
	protected void configureJob(JobConf conf){
		super.configureJob(conf, false);
	}
}

class DumpTableTrivialReducer{
	
}